package com.kancy.streamplus.channel;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.AbstractMap;
import java.util.Map;
import java.util.TreeMap;


/**
 * PollingMessageDispatcher
 *
 * @author kancy
 * @date 2020/3/22 15:54
 */
public class PollingMessageDispatcher {
    private static final Logger log = LoggerFactory.getLogger(PollingMessageDispatcher.class);

    private final Map<Long, PeriodChannel> period = new TreeMap();

    private final ApplicationContext context;
    private final DynamicChannelResolver dynamicChannelResolver;

    public PollingMessageDispatcher(ApplicationContext context, DynamicChannelResolver dynamicChannelResolver) {
        this.context = context;
        this.dynamicChannelResolver = dynamicChannelResolver;
    }

    /**
     * 监听消息
     * @param message
     */
    @StreamListener(PollingInputChannel.INPUT)
    public void onMessage(Message<String> message) {
        try {
            processMessage(message);
        } catch (Exception e) {
            log.error(String.format("PollingMessageDispatcher 消费转发消息失败：%s", message), e);
            throw e;
        }
    }

    /**
     * 处理消息
     * @param message
     */
    private void processMessage(Message<String> message) {
        MessageHeaders messageHeaders = message.getHeaders();
        long channelKey = getChannelKey(calcNextOperateTime(messageHeaders));
        if (channelKey > 0L) {
            log.info("转移需要等待的消息至[{}]延时队列, channel:{} , messageType:{}", channelKey,
                    messageHeaders.get(PollingMessage.CHANNEL_NAME), messageHeaders.get(PollingMessage.MESSAGE_TYPE));
            period.get(channelKey).output().send(message);
        } else {
            log.info("转移消息至channel:{} , messageType:{} , content:{}",
                    messageHeaders.get(PollingMessage.CHANNEL_NAME), messageHeaders.get(PollingMessage.MESSAGE_TYPE),
                    message.getPayload());
            dynamicChannelResolver.getChannel(messageHeaders.get(PollingMessage.CHANNEL_NAME, String.class)).send(message);
        }
    }

    /**
     * 获取等待队列的Key并写入period map，按从大到小排序
     *
     * @param nextTrackingTime 下个跟踪周期的时间
     * @return channelKey
     */
    private long getChannelKey(Long nextTrackingTime) {
        final long timeToNextTracking = nextTrackingTime - System.currentTimeMillis();
        if (timeToNextTracking < 0) {
            return 0;
        }
        return period.entrySet().stream()
                .sorted((x, y) -> (int) (y.getKey() - x.getKey()))
                .filter(x -> x.getKey() <= timeToNextTracking)
                .findFirst()
                .orElse(new AbstractMap.SimpleImmutableEntry<>(0L, null))
                .getKey();
    }

    /**
     * 计算下一次操作时间
     * @param messageHeaders
     * @return
     */
    private long calcNextOperateTime(MessageHeaders messageHeaders) {
        Long delay = messageHeaders.get(PollingMessage.DELAY, Long.class);
        Long lastQueryTimestamp = messageHeaders.get(PollingMessage.LAST_QUERY_TIMESTAMP, Long.class);
        return lastQueryTimestamp + delay;
    }


    @PostConstruct
    public void initPeriodMap() {
        Map<String, PeriodChannel> channelMap = context.getBeansOfType(PeriodChannel.class);
        channelMap.entrySet().stream().forEach(e ->{
            String channelClassName = e.getKey();
            try {
                Class<?> channelClass = Class.forName(channelClassName);
                if (channelClass.isAnnotationPresent(Period.class)){
                    Period annotation = channelClass.getDeclaredAnnotation(Period.class);
                    Duration duration = Duration.parse(annotation.value());
                    long periodTime = duration.toMillis();
                    if (periodTime > 0){
                        PeriodChannel periodChannel = e.getValue();
                        period.put(periodTime, periodChannel);
                    }else {
                        log.warn("PeriodChannel [{}] period value less than zero", channelClassName);
                    }
                }
            } catch (Exception ex) {
                throw new RuntimeException(ex);
            }
        });

        if (period.isEmpty()){
            log.warn("Not find PeriodChannel.");
        }
    }
}
